package com.wulaobo.partition;

import com.wulaobo.constant.Constants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Author wuwenbo
 * @Date 2024/5/24 9:54
 * @ClassName: PartitionCustomDemo
 * @Description: TODO
 */
public class PartitionCustomDemo {

    public static void main(String[] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        env.setParallelism(10);

        DataStreamSource<String> socketDS = env.socketTextStream(Constants.IP, Constants.SOCKET_PORT);

        DataStream<String> myDS = socketDS
                .partitionCustom(
                        new MyPartitioner(),
                        value -> value);

        myDS.print();

        env.execute();
    }

}
